package com.it.operator;

import com.it.pojo.Event;
import com.it.operator.utils.SourceUtils;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Operator_KeyBY {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment);
        //max：只更改我们传入的列为最大值，其他列是该key是第一次出现的值：Event{user='tom', url='./index', timestamp=1970-01-01 08:00:40.0}
        eventSource.keyBy((KeySelector<Event, String>) value -> value.user).max("timestamp").print("max:");
        ////maxBy：按照传入列，找到了列最大的那条记录：Event{user='tom', url='./order?id=10', timestamp=1970-01-01 08:00:40.0}
        eventSource.keyBy((KeySelector<Event, String>) value -> value.user).maxBy("timestamp").print("maxBy:");
        executionEnvironment.execute();

    }
}
